package com.jd.platform.gobrs.dashboard.netty;

import cn.hutool.core.util.StrUtil;
import com.github.benmanes.caffeine.cache.Cache;
import com.jd.platform.gobrs.common.model.GCacheKeyModel;
import com.jd.platform.gobrs.common.rule.KeyRule;
import com.jd.platform.gobrs.dashboard.cache.CaffeineBuilder;
import com.jd.platform.gobrs.dashboard.common.domain.req.SearchReq;
import com.jd.platform.gobrs.dashboard.model.KeyTimely;
import com.jd.platform.gobrs.dashboard.util.RuleUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * 此处存储所有发来的热key，统一处理入库
 *
 * @author sizegang
 * @version 1.0
 * @date 2020-08-31
 */
public class AutoKeyReceiver {

    /**
     * 自动key 只接收前一万个 做数据缓冲
     */
    private static LinkedBlockingQueue<GCacheKeyModel> autoKeyReceiverQueue = new LinkedBlockingQueue<GCacheKeyModel>();
    /**
     * 存储实时Autokey，供界面查询实时热key
     */
    private static Map<Integer, Cache<String, Object>> aliveKeyStore = new ConcurrentHashMap<>();

    private static Logger logger = LoggerFactory.getLogger("AutoKeyReceiver");

    private static Integer defaultKeyDuration = -1;

    /**
     * netty收到的先存这里
     */
    public static void push(GCacheKeyModel model) {
        boolean offer = autoKeyReceiverQueue.offer(model);
        if (!offer) {
            logger.warn("auto Key Queue is full dont receive key...");
        }
    }

    public static GCacheKeyModel take() {
        try {
            return autoKeyReceiverQueue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 将热key存入本地缓存，设置过期时间
     */
    public static void writeToLocalCaffeine(GCacheKeyModel autoKey) {
        String appNameKey = autoKey.getAppName() + "/" + autoKey.getKey();
        KeyRule keyRule = RuleUtil.findByKey(appNameKey);
        Cache<String, Object> cache;
        if (keyRule == null) {
            cache = aliveKeyStore.computeIfAbsent(defaultKeyDuration, s -> CaffeineBuilder.cache(defaultKeyDuration));
        } else {
            cache = aliveKeyStore.computeIfAbsent(keyRule.getDuration(), s -> CaffeineBuilder.cache(keyRule.getDuration()));
        }
        cache.put(appNameKey, autoKey);
    }

    /**
     * 展示当前所有的实时热key
     */
    public static List<KeyTimely> list(SearchReq searchReq) {
        List<KeyTimely> timelyList = new ArrayList<>();

        long now = System.currentTimeMillis();

        for (Integer duration : aliveKeyStore.keySet()) {
            Cache<String, Object> cache = aliveKeyStore.get(duration);
            ConcurrentMap<String, Object> concurrentHashMap = cache.asMap();
            for (Map.Entry<String, Object> entry : concurrentHashMap.entrySet()) {
                KeyTimely keyTimely = parse((GCacheKeyModel) entry.getValue(), now);
                if (keyTimely == null) {
                    continue;
                }
                timelyList.add(keyTimely);
            }
        }

        if (searchReq != null) {
            if (StrUtil.isNotEmpty(searchReq.getApp())) {
                timelyList = timelyList.parallelStream().filter(keyTimely -> searchReq.getApp().equals(keyTimely.getAppName())).collect(Collectors.toList());
            }
            if (StrUtil.isNotEmpty(searchReq.getKey())) {
                timelyList = timelyList.parallelStream().filter(keyTimely -> keyTimely.getKey().startsWith(searchReq.getKey())).collect(Collectors.toList());
            }

        }

        timelyList.sort(Comparator.comparing(KeyTimely::getCreateTime).reversed());

        return timelyList;
    }

    /**
     * 将hotkeyModel变成前端需要的对象
     */
    private static KeyTimely parse(GCacheKeyModel hotKeyModel, long now) {
        String appNameKey = hotKeyModel.getAppName() + "/" + hotKeyModel.getKey();
        KeyRule keyRule = RuleUtil.findByKey(appNameKey);
        if (keyRule == null) {
            return null;
        }
        long remainTime = keyRule.getDuration() * 1000 - (now - hotKeyModel.getCreateTime());
        return KeyTimely.aKeyTimely()
                .key(hotKeyModel.getKey())
                .val(UUID.randomUUID().toString())
                .appName(hotKeyModel.getAppName())
                .duration(remainTime / 1000)
                .ruleDesc(RuleUtil.ruleDesc(appNameKey))
                .createTime(new Date(hotKeyModel.getCreateTime())).build();
    }

    /**
     * 删除实时热key
     */
    public static boolean delete(String appNameKey) {
        KeyRule keyRule = RuleUtil.findByKey(appNameKey);
        if (keyRule == null) {
            return false;
        }
        Cache<String, Object> cache = aliveKeyStore.get(keyRule.getDuration());
        if (cache == null) {
            return false;
        }
        cache.invalidate(appNameKey);
        return true;
    }

    /**
     * 定时清理caffeine
     */
    public static void cleanUpCaffeine() {
        for (Integer duration : aliveKeyStore.keySet()) {
            Cache<String, Object> cache = aliveKeyStore.get(duration);
            cache.cleanUp();
        }
    }

}
